package com.shanlitech.et;

import com.google.protobuf.Descriptors;
import com.google.protobuf.DynamicMessage;
import com.shanlitech.et.pb.ZProtobuf;
import com.shanlitech.et.pb.Agpub;
import org.zeromq.ZActor;
import org.zeromq.ZContext;
import org.zeromq.ZMQ;
import org.zeromq.ZMQ.Socket;
import org.zeromq.ZPoller;

import java.util.*;

public class SpeechRecorder extends ZActor.SimpleActor {

	public static class SpeechInfo {
		public String		ag_id;
		public Long			speech_id;
		public Long			gid;
		public Integer		uid;
		public Long			t_start;
		public Long			t_end;
		public Integer		payload;

		public SpeechInfo() {
			ag_id = new String();
		}
	}
	
	private static class SingletonHolder {

        private static final SpeechRecorder INSTANCE = new SpeechRecorder("tcp://192.168.0.222:6041");

    }

    public static final SpeechRecorder getInstance() {

        return SingletonHolder.INSTANCE;

    }
	
	

	public static class SpeechRecord {
		SpeechInfo				info;
		public List<byte[]>		frames;

		public SpeechRecord() {
			info = new SpeechInfo();
			frames = new LinkedList<byte[]>();
		}
	}



	private ZActor								zactor;
	public ZProtobuf zpb;
	private String								ag_address;
	public Map<Long,LinkedList<SpeechRecord>>	group_records;
	public Map<Long,SpeechRecord>				speech_records;
	public Map<Long,SpeechRecord>				records_uncompleted;

	public Descriptors.FieldDescriptor			fd_gotmic_ag_id;
	public Descriptors.FieldDescriptor			fd_gotmic_speech_id;
	public Descriptors.FieldDescriptor			fd_gotmic_gid;
	public Descriptors.FieldDescriptor			fd_gotmic_uid;
	public Descriptors.FieldDescriptor			fd_gotmic_payload;

	public Descriptors.FieldDescriptor			fd_lostmic_ag_id;
	public Descriptors.FieldDescriptor			fd_lostmic_speech_id;
	public Descriptors.FieldDescriptor			fd_lostmic_gid;
	public Descriptors.FieldDescriptor			fd_lostmic_uid;
	public Descriptors.FieldDescriptor			fd_lostmic_payload;

	public Descriptors.FieldDescriptor			fd_ap_ag_id;
	public Descriptors.FieldDescriptor			fd_ap_speech_id;
	public Descriptors.FieldDescriptor			fd_ap_gid;
	public Descriptors.FieldDescriptor			fd_ap_uid;
	public Descriptors.FieldDescriptor			fd_ap_payload;
	public Descriptors.FieldDescriptor			fd_ap_seq;
	public Descriptors.FieldDescriptor			fd_ap_ts;
	public Descriptors.FieldDescriptor			fd_ap_data;
	
	
	public Map<Long,LinkedList<SpeechRecord>> get_group_records(){
		return group_records;
	} 

	private void loadDescriptors() {
		Descriptors.Descriptor md;

	    md = Agpub.MemberGotMic.getDescriptor();
		fd_gotmic_ag_id = md.findFieldByName("ag_id");
		fd_gotmic_speech_id = md.findFieldByName("speech_id");
		fd_gotmic_gid = md.findFieldByName("gid");
		fd_gotmic_uid = md.findFieldByName("uid");
		fd_gotmic_payload = md.findFieldByName("payload");

		md = Agpub.MemberLostMic.getDescriptor();
		fd_lostmic_ag_id = md.findFieldByName("ag_id");
		fd_lostmic_speech_id = md.findFieldByName("speech_id");
		fd_lostmic_gid = md.findFieldByName("gid");
		fd_lostmic_uid = md.findFieldByName("uid");
		fd_lostmic_payload = md.findFieldByName("payload");


		md = Agpub.AudioPacket.getDescriptor();
		fd_ap_ag_id = md.findFieldByName("ag_id");
		fd_ap_speech_id = md.findFieldByName("speech_id");
		fd_ap_gid = md.findFieldByName("gid");
		fd_ap_uid = md.findFieldByName("uid");
		fd_ap_payload = md.findFieldByName("payload");
		fd_ap_seq = md.findFieldByName("seq");
		fd_ap_ts = md.findFieldByName("timestamp");
		fd_ap_data = md.findFieldByName("data");
	}

	public SpeechRecorder(String audiogate_address) {
		zactor = null;
		zpb = new ZProtobuf(Agpub.getDescriptor());
		ag_address = audiogate_address;
		group_records = new HashMap<Long,LinkedList<SpeechRecord>>();
		speech_records = new HashMap<Long,SpeechRecord>();
		records_uncompleted = new HashMap<Long,SpeechRecord>();
		loadDescriptors();
	}

	public boolean start() {
		if( zactor == null ) {
			zactor = new ZActor(this,"SpeechRecorder");
			return true;
		}
		return false;
	}

	public void stop() {
		if( zactor != null ) {
			Socket pipe = zactor.pipe();
			pipe.send("$TERM");
			zactor = null;
		}
	}

	public List<SpeechInfo> findSpeechInfoByGid(long gid) {
		List<SpeechInfo> results = null;
		synchronized(this) {
			List<SpeechRecord> rs = group_records.get(gid);
			if( rs != null ) {
				results = new ArrayList<SpeechInfo>();
				Iterator it = rs.iterator();
				while( it.hasNext() ) {
					results.add( ((SpeechRecord)it.next()).info );
				}
			}
		}
		return results;
	}

	public SpeechRecord findSpeechRecordById(long speech_id) {
		SpeechRecord r = null;
		synchronized(this) {
			r = speech_records.get(speech_id);
		}
		return r;
	}

	@Override
	public List<Socket> createSockets(final ZContext ctx, final Object... args) {
		ZMQ.Socket pub = ctx.createSocket(ZMQ.PUB);

		pub.bind("tcp://*:6041");
		//logger.info("建立连接成功: " + ag_address);
		return Arrays.asList(pub);
	}

	@Override
    public void start(Socket pipe, List<Socket> sockets, ZPoller poller) {
		for(Socket s : sockets) {
			poller.register(s, ZPoller.POLLIN);
			//logger.info("start..........");
		}
	}

	@Override
	public boolean backstage(Socket pipe, ZPoller poller, int events) {
		String cmd = pipe.recvStr();
		if ("$TERM".equals(cmd)) {
			// end of the actor
			return false;
		}
		return true;
	}

	@Override
	public boolean stage(Socket socket, Socket pipe, ZPoller poller, int events) {
		ZProtobuf.RecvStruct pkg;
		pkg = zpb.recv(socket,false);
		if( pkg != null && pkg.msg != null ) {
			Descriptors.Descriptor md = pkg.msg.getDescriptorForType();
			//logger.info( "recv pub message: " + md.getFullName());
			
			if( md.equals(Agpub.AudioPacket.getDescriptor()) ) {
				onAudioPacket(md,pkg.msg);
			} else if( md.equals(Agpub.MemberGotMic.getDescriptor()) ) {
				onMemberGotMic(md,pkg.msg);
			} else if( md.equals(Agpub.MemberLostMic.getDescriptor()) ) {
				onMemberLostMic(md,pkg.msg);
			} 
		} else {
			//logger.info("fail to recv pkg");
		}
		return true;
	}

	private void onMemberGotMic(Descriptors.Descriptor md,DynamicMessage msg) {
		SpeechRecord record = new SpeechRecord();
		do {
			Object v;
			v = msg.getField(fd_gotmic_ag_id);
			if( v == null ) break;
			record.info.ag_id = (String)v;
			
			v = msg.getField(fd_gotmic_speech_id);
			if( v == null ) break;
			record.info.speech_id = (Long)v;

			v = msg.getField(fd_gotmic_gid);
			if( v == null ) break;
			record.info.gid = (Long)v;

			v = msg.getField(fd_gotmic_uid);
			if( v == null ) break;
			record.info.uid = (Integer)v;

			v = msg.getField(fd_gotmic_payload);
			if( v == null ) break;
			record.info.payload = (Integer)v;
			
			record.info.t_start=System.currentTimeMillis();
			
			records_uncompleted.put(record.info.speech_id,record);
		} while(false);
	}

	private void onAudioPacket(Descriptors.Descriptor md,DynamicMessage msg) {
		do {
			Object v;
			v = msg.getField(fd_ap_speech_id);
			if( v == null ) {
				// warnning
				break;
			}

			Long speech_id = (Long)v;
			SpeechRecord record = records_uncompleted.get(speech_id);
			if( record == null ) {
				// warnning
				break;
			}

			v = msg.getField(fd_ap_data);
			if( v == null ) break;
			com.google.protobuf.ByteString bs = (com.google.protobuf.ByteString)v;
			record.frames.add(bs.toByteArray());
		} while(false);
	}

	private void onMemberLostMic(Descriptors.Descriptor md,DynamicMessage msg) {
		do {
			Object v;
			v = msg.getField(fd_lostmic_speech_id);
			if( v == null ) {
				// warnning
				break;
			}

			Long speech_id = (Long)v;
			SpeechRecord record = records_uncompleted.get(speech_id);
			if( record == null ) {
				// warnning
				break;
			}
			
			records_uncompleted.remove(speech_id);
			synchronized(this) {
				
				LinkedList<SpeechRecord> rl = group_records.get(record.info.gid);
				if( rl == null ) {
					rl = new LinkedList<SpeechRecord>();
					group_records.put(record.info.gid,rl);
				} else if( rl.size() >= 3 ) {
					SpeechRecord expired = rl.removeFirst();
					speech_records.remove(expired.info.speech_id);
				}
				record.info.t_end=System.currentTimeMillis();
				rl.add(record);
				speech_records.put(record.info.speech_id,record);
			}
			System.out.println("finish a speech");
		} while(false);
	}

	@Override
	public void closed(Socket socket) {
		super.closed(socket);
		System.out.println("socket连接关闭");
	}

	@Override
	public boolean destroyed(ZContext ctx, Socket pipe, ZPoller poller) {
		System.out.println("socket连接销毁");
		return super.destroyed(ctx, pipe, poller);
	}

	@Override
	public boolean finished(Socket pipe) {
		System.out.println("socket连接完成");
		return super.finished(pipe);
	}
}
